package com.xiaoying.server.publish.impl;

import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.jms.support.converter.MessageConversionException;
import org.springframework.stereotype.Component;

import com.xiaoying.server.publish.entity.FileInfo;
import com.xiaoying.server.publish.entity.UploadResult;
import com.xiaoying.server.publish.utils.DateUtil;
import com.xiaoying.server.publish.utils.ObjectMessageConverter;
import com.xiaoying.server.publish.utils.RedisUtil;
import com.xiaoying.server.publish.utils.SerializeUtil;

public class FetchRequest implements Runnable {
    private static Log log = LogFactory.getFactory().getInstance(FetchRequest.class);
    
    private  int maxRecord = 50;	//内存队列中最大文件数量
    private int DELAY = 1000;
    private int MAXDELAYCOUNT = 60;	//没有待传文件请求时，最长暂停时间(单位秒)
    private boolean isOpen = true;
    private BlockingQueue<FileInfo> fileUploadQueue ;
    private BlockingQueue<UploadResult> uploadResultQueue ;
    
    // 消息和对象的转换类
	private ObjectMessageConverter omc ;		
    // 连接工厂,JMS用它创建连接
	private ConnectionFactory connectionFactory;
    // 客户端到JMS Provider的连接
	private Connection connection;
    // 发送或接收消息的线程
	private Session session;
    // 消息的目的地(消息发送给谁)
	private Destination destination;
    // 消费者,消息接收者
	private MessageConsumer consumer;
    
    
    public FetchRequest(BlockingQueue<FileInfo> reqQueue,BlockingQueue<UploadResult> resQueue){
        this.fileUploadQueue = reqQueue;
        this.uploadResultQueue = resQueue;
        
        this.connectionFactory = new ActiveMQConnectionFactory(
	            ActiveMQConnection.DEFAULT_USER,
	            ActiveMQConnection.DEFAULT_PASSWORD,
	            "tcp://localhost:61616");
        this.omc= new ObjectMessageConverter();	    
        try {
        	// 构造从工厂得到连接对象
			this.connection = connectionFactory.createConnection();
	        // 启动连接
	        this.connection.start();
	        // 获取session
	        this.session = connection.createSession(Boolean.FALSE,
	                Session.AUTO_ACKNOWLEDGE);
	        this.destination = session.createQueue("xiaoying");
	        this.consumer = session.createConsumer(destination);
		} catch (JMSException e) {
			log.error(e);
		}                
        Thread proc = new Thread(this);
        proc.start();
    }
    
    @Override
    public synchronized void run() {
    	int i =0;
    	while(true){
    		FileInfo fileInfo;
			try {				
				if(fileUploadQueue.size()<maxRecord &&  uploadResultQueue.size()<maxRecord){
					fileInfo = (FileInfo) omc.fromMessage(consumer.receive());					
		            if (null != fileInfo) {
		            	
		            	fileInfo.setReceiveDt(DateUtil.getDateTimeByFormat(new Date()));
		            	fileUploadQueue.add(fileInfo);		            	
		            	// 接收的到的消息放入redis中备份
		            	this.put2Redis(fileInfo);      	
		            	log.debug("收到1条消息: " + fileInfo.getId());
		            } else {
		            	log.debug("接收消息失败!");
		            }
				}	//end if
				if(fileUploadQueue.size()<=0){
	              if (i<MAXDELAYCOUNT) {
	                    i++;
	                }
				}else{
					i = 1;
				}
				// 线程wait一段时间
				wait(i*DELAY);	
			} catch (MessageConversionException e) {
				log.error("MQ消息转换成对象失败 "+e);
			} catch (JMSException e) {
				log.error("MQ消息接收失败  "+e);
			} catch (InterruptedException e) {
				log.error(e);
			} 
    	}
    }

    /**
     * 接收的到的消息放入redis中备份
     * @author ZJD
     * @param fileInfo
     */
    private void put2Redis(FileInfo fileInfo){
		String key = RedisUtil.PREFIX_XY_BACKUP_UPLOAD_FILEINFO+fileInfo.getFileName()+"_"+fileInfo.getReceiveDt();
		Map<String, String> map = new HashMap<String, String>();
		map.put("id", String.valueOf(fileInfo.getId()));
		map.put("fileName", fileInfo.getFileName());
		map.put("filePath", fileInfo.getFilePath());
		map.put("fileSize", String.valueOf(fileInfo.getFileSize()));    		
		map.put("fileSuffix", fileInfo.getFileSuffix());
		map.put("fileType", String.valueOf(fileInfo.getFileType()));
		map.put("uploadCount", String.valueOf(fileInfo.getUploadCount()));
		map.put("receiveDt", fileInfo.getReceiveDt());		    		
		RedisUtil.getRedisInstance().hmset(key, map);
    }
    
    
    
    public void ShutdownHook() throws Exception {
        Runtime.getRuntime().addShutdownHook(new Thread() {
            public void run() {
                log.info("FetchRequest shutdown.");
            }
        });
    }
    
}
